package org.atlantis.datasink.server.buildin;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.atlantis.datasink.codec.CodecOption;
import org.atlantis.datasink.codec.CommandFactory;
import org.jboss.netty.channel.Channel;

class CommandSender {

	private ScheduledExecutorService executor = Executors.newScheduledThreadPool(3);
	private ConcurrentMap<SocketAddress, ScheduledFuture<?>> commandMap = new ConcurrentHashMap<SocketAddress, ScheduledFuture<?>>();

	void registerCommand(CodecOption codecOpt, Channel channel) {
		final CommandFactory commandFactory = codecOpt.getRequestFactory();
		final Channel _c = channel;
		ScheduledFuture<?> future = executor.scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				if (!_c.isConnected()) return;
				_c.write(commandFactory.createCommand(_c));
			}
		}, 5000L, codecOpt.getCommandSendInterval(), TimeUnit.MILLISECONDS);
		commandMap.putIfAbsent(_c.getRemoteAddress(), future);
	}

	void unregisterCommand(SocketAddress addr) {
		ScheduledFuture<?> future = commandMap.get(addr);
		if (future == null) return;
		if (future.cancel(true)) commandMap.remove(addr);
	}

	void destory() {
		for (ScheduledFuture<?> future : commandMap.values())
			future.cancel(true);
		executor.shutdownNow();
		commandMap.clear();
		commandMap = null;
	}
}
